package com.yvan.elasticjob;

import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.event.rdb.JobEventRdbConfiguration;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.lite.spring.api.SpringJobScheduler;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperConfiguration;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
import com.google.common.collect.Maps;
import com.yvan.leto.boot.LetoSpringContext;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeansException;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import javax.sql.DataSource;
import java.util.HashSet;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

@Configuration
@EnableConfigurationProperties(ElasticJobProperties.class)
@ConditionalOnBean(annotation = EnableElasticJob.class)
public class ElasticJobAutoConfiguration implements ApplicationContextAware {

    private final ElasticJobProperties elasticJobProperties;
    private ApplicationContext applicationContext;

    private Map<String, Set<BeanJobSchedulerHolder>> beanJobSchedulerSubscriptions = Maps.newHashMap();

    public ElasticJobAutoConfiguration(ElasticJobProperties elasticJobProperties) {
        this.elasticJobProperties = elasticJobProperties;
        LetoSpringContext.setElasticJobAutoConfiguration(this);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @PostConstruct
    public void initElasticJob() {
        ZookeeperRegistryCenter regCenter = new ZookeeperRegistryCenter(new ZookeeperConfiguration(elasticJobProperties.getServerList(), elasticJobProperties.getNamespace()));
        regCenter.init();
        Map<String, SimpleJob> map = applicationContext.getBeansOfType(SimpleJob.class);

        for (Map.Entry<String, SimpleJob> entry : map.entrySet()) {
            SimpleJob simpleJob = entry.getValue();
            setElasticSimpleJob(regCenter,simpleJob);
            setReloadElasticSimpleJob(regCenter,simpleJob);
        }
    }

    public void updateCronOnBean() {
        for (Set<BeanJobSchedulerHolder> beanJobSchedulerHolders:beanJobSchedulerSubscriptions.values()){
            for(BeanJobSchedulerHolder beanJobSchedulerHolder:beanJobSchedulerHolders){
                updateReloadableElasticSimpleCron(beanJobSchedulerHolder);
            }
        }
    }

    private void setElasticSimpleJob(ZookeeperRegistryCenter regCenter,SimpleJob simpleJob){
        ElasticSimpleJob elasticSimpleJobAnnotation = simpleJob.getClass().getAnnotation(ElasticSimpleJob.class);
        if(null == elasticSimpleJobAnnotation)
            return;

        String cron = StringUtils.defaultIfBlank(elasticSimpleJobAnnotation.cron(), elasticSimpleJobAnnotation.value());
        SimpleJobConfiguration simpleJobConfiguration = new SimpleJobConfiguration(JobCoreConfiguration.newBuilder(simpleJob.getClass().getName(), cron, elasticSimpleJobAnnotation.shardingTotalCount()).shardingItemParameters(elasticSimpleJobAnnotation.shardingItemParameters()).build(), simpleJob.getClass().getCanonicalName());
        LiteJobConfiguration liteJobConfiguration = LiteJobConfiguration.newBuilder(simpleJobConfiguration).overwrite(true).build();

        String dataSourceRef = elasticSimpleJobAnnotation.dataSource();
        if (StringUtils.isNotBlank(dataSourceRef)) {

            if (!applicationContext.containsBean(dataSourceRef)) {
                throw new RuntimeException("not exist datasource [" + dataSourceRef + "] !");
            }

            DataSource dataSource = (DataSource) applicationContext.getBean(dataSourceRef);
            JobEventRdbConfiguration jobEventRdbConfiguration = new JobEventRdbConfiguration(dataSource);
            SpringJobScheduler jobScheduler = new SpringJobScheduler(simpleJob, regCenter, liteJobConfiguration, jobEventRdbConfiguration);
            jobScheduler.init();
        } else {
            SpringJobScheduler jobScheduler = new SpringJobScheduler(simpleJob, regCenter, liteJobConfiguration);
            jobScheduler.init();
        }
    }

    private void setReloadElasticSimpleJob(ZookeeperRegistryCenter regCenter,SimpleJob simpleJob){
        ReloadableElasticSimpleCron elasticSimpleJobAnnotation = simpleJob.getClass().getAnnotation(ReloadableElasticSimpleCron.class);
        if(null == elasticSimpleJobAnnotation)
            return;

        final Object property = getProperties().get(elasticSimpleJobAnnotation.value());
        String cron = StringUtils.defaultIfBlank(property.toString(), elasticSimpleJobAnnotation.value());
        SimpleJobConfiguration simpleJobConfiguration = new SimpleJobConfiguration(JobCoreConfiguration.newBuilder(simpleJob.getClass().getName(), cron, elasticSimpleJobAnnotation.shardingTotalCount()).shardingItemParameters(elasticSimpleJobAnnotation.shardingItemParameters()).build(), simpleJob.getClass().getCanonicalName());
        LiteJobConfiguration liteJobConfiguration = LiteJobConfiguration.newBuilder(simpleJobConfiguration).overwrite(true).build();

        String dataSourceRef = elasticSimpleJobAnnotation.dataSource();
        SpringJobScheduler jobScheduler;
        if (StringUtils.isNotBlank(dataSourceRef)) {

            if (!applicationContext.containsBean(dataSourceRef)) {
                throw new RuntimeException("not exist datasource [" + dataSourceRef + "] !");
            }

            DataSource dataSource = (DataSource) applicationContext.getBean(dataSourceRef);
            JobEventRdbConfiguration jobEventRdbConfiguration = new JobEventRdbConfiguration(dataSource);
            jobScheduler = new SpringJobScheduler(simpleJob, regCenter, liteJobConfiguration, jobEventRdbConfiguration);
        } else {
            jobScheduler = new SpringJobScheduler(simpleJob, regCenter, liteJobConfiguration);
        }

        jobScheduler.init();

        if (!beanJobSchedulerSubscriptions.containsKey(elasticSimpleJobAnnotation.value())) {
            beanJobSchedulerSubscriptions.put(elasticSimpleJobAnnotation.value(), new HashSet<BeanJobSchedulerHolder>());
        }
        beanJobSchedulerSubscriptions.get(elasticSimpleJobAnnotation.value()).add(new BeanJobSchedulerHolder(simpleJob,jobScheduler));
    }

    private void updateReloadableElasticSimpleCron(BeanJobSchedulerHolder beanJobSchedulerHolder){

        ReloadableElasticSimpleCron elasticSimpleJobAnnotation = beanJobSchedulerHolder.getBean().getClass().getAnnotation(ReloadableElasticSimpleCron.class);

        final Object property = getProperties().get(elasticSimpleJobAnnotation.value());
        String cron = StringUtils.defaultIfBlank(property.toString(), elasticSimpleJobAnnotation.value());
        SimpleJobConfiguration simpleJobConfiguration = new SimpleJobConfiguration(JobCoreConfiguration.newBuilder(
                beanJobSchedulerHolder.getBean().getClass().getName(),
                cron,
                elasticSimpleJobAnnotation.shardingTotalCount()).shardingItemParameters(elasticSimpleJobAnnotation.shardingItemParameters()).build(),
                beanJobSchedulerHolder.getBean().getClass().getCanonicalName());
        LiteJobConfiguration liteJobConfiguration = LiteJobConfiguration.newBuilder(simpleJobConfiguration).overwrite(true).build();

        beanJobSchedulerHolder.getJobScheduler().getSchedulerFacade().updateJobConfiguration(liteJobConfiguration);
    }

    private Properties getProperties() {
        return LetoSpringContext.getReloadProperties();
    }
}
